package com.atguigu.wuliu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.wuliu.common.WuliuConfig;
import com.atguigu.wuliu.utils.HbaseUtil;
import com.atguigu.wuliu.utils.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;


//异步方法sku关联维度数据
public abstract class DimAsyncByForeignKeyFunction<T> extends RichAsyncFunction<T,T> implements DimJoinForeignFunction<T> {
    private String tableName;
    private StatefulRedisConnection<String,String> asyncRedisConn;
    private AsyncConnection asyncHbaseConn;

    public DimAsyncByForeignKeyFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        asyncRedisConn = RedisUtil.getAsyncRedisConnection();
        asyncHbaseConn = HbaseUtil.getAsyncConnection();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeAsyncRedisConnection(asyncRedisConn);
        HbaseUtil.closeAsyncConnection(asyncHbaseConn);
    }
    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        CompletableFuture.supplyAsync(
                //开始一个线程任务，有返回值
                new Supplier<JSONObject>() {
                    @Override
                    public JSONObject get() {
                        //发送异步请求，从Redis中获取维度数据
                        JSONObject dimJsonObj = RedisUtil.asyncGetDimInfo(asyncRedisConn, tableName + ":" + getForeignKey(obj).f1);
                        return dimJsonObj;
                    }
                }
        ).thenApplyAsync(
                //发送异步请求，并将上一个线程任务返回结果作为当前任务的参数，并有返回值
                new Function<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject apply(JSONObject dimJsonObj) {
                        if(dimJsonObj == null){
                            //说明没有从redis中获取到维度数据，发送异步请求到hbase查询维度

                            dimJsonObj = HbaseUtil.getRowByForeignKey(asyncHbaseConn, WuliuConfig.HBASE_NAMESPACE, tableName, getForeignKey(obj));
                            if(dimJsonObj != null){
                                System.out.println(tableName + "  " + getForeignKey(obj) + "  从 hbase 读取");
                                //将从hbase中查询到的维度数据 放到redis中缓存起来
                                RedisUtil.asyncWriteDim(asyncRedisConn,tableName + ":" + getForeignKey(obj).f1,dimJsonObj);
                            }
                        }else{
                            //说明从redis中获取到维度数据
                            System.out.println(tableName + "  " + getForeignKey(obj).f1 + "  从 redis 读取");
                        }
                        return dimJsonObj;
                    }
                }
        ).thenAcceptAsync(
                //发送异步请求，并将上一个线程任务返回结果作为当前任务的参数，没有有返回值
                new Consumer<JSONObject>() {
                    @Override
                    public void accept(JSONObject dimJsonObj) {
                        if(dimJsonObj != null){
                            //将维度对象相关的维度属性补充到流中的对象上
                            join(obj,dimJsonObj);
                            //向下游传递数据
                            resultFuture.complete(Collections.singleton(obj));
                        }
                    }
                });
    }
    // 如果异步超时,会执行这个方法
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        // 根据自己情况进行处理
        throw new RuntimeException("``````````````````` 异步超时~````````````````");
    }



}
